package net.flyinggroup.spring;

import lombok.SneakyThrows;
import lombok.extern.slf4j.Slf4j;
import org.junit.jupiter.api.Test;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Hooks;
import reactor.core.publisher.Mono;
import reactor.core.scheduler.Schedulers;

import java.time.Duration;
import java.time.LocalTime;
import java.util.Arrays;
import java.util.List;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.stream.Collectors;
import java.util.stream.IntStream;

@Slf4j
public class Debugging {
    @Test
    public void imperative() throws InterruptedException, ExecutionException {
        final ScheduledExecutorService executor =
                Executors.newSingleThreadScheduledExecutor();

        int seconds = LocalTime.now().getSecond();
        List<Integer> source;
        if (seconds % 2 == 0) {
            source = IntStream.range(1, 11).boxed().collect(Collectors.toList());
        } else if (seconds % 3 == 0) {
            source = IntStream.range(0, 4).boxed().collect(Collectors.toList());
        } else {
            source = Arrays.asList(1, 2, 3, 4);
        }

        executor.submit(() -> source.get(5))  //line 76
                .get();
    }

    @Test
    public void reactive() {
        int seconds = LocalTime.now().getSecond();
        Mono<Integer> source;
        if (seconds % 2 == 0) {
            source = Flux.range(1, 10)
                    .elementAt(5);
        } else if (seconds % 3 == 0) {
            source = Flux.range(0, 4)
                    .elementAt(5);
        } else {
            source = Flux.just(1, 2, 3, 4)
                    .elementAt(5);
        }
        source.subscribeOn(Schedulers.parallel()).block(); //line 97
    }

    @Test
    public void log() {
        int seconds = LocalTime.now().getSecond();
        Mono<Integer> source;
        if (seconds % 2 == 0) {
            source = Flux.range(1, 10)
                    .elementAt(5)
                    .log("source A");
        } else if (seconds % 3 == 0) {
            source = Flux.range(0, 4)
                    .elementAt(5)
                    .log("source B");
        } else {
            source = Flux.just(1, 2, 3, 4)
                    .elementAt(5)
                    .log("source C");
        }

        source.block(); //line 138
    }

    @Test
    public void hook() {
        Hooks.onOperatorDebug();
        try {
            int seconds = LocalTime.now().getSecond();
            Mono<Integer> source;
            if (seconds % 2 == 0) {
                source = Flux.range(1, 10)
                        .elementAt(5); //line 149
            } else if (seconds % 3 == 0) {
                source = Flux.range(0, 4)
                        .elementAt(5); //line 153
            } else {
                source = Flux.just(1, 2, 3, 4)
                        .elementAt(5); //line 157
            }

            source.block(); //line 160
        } finally {
            Hooks.resetOnOperatorDebug();
        }
    }

    @Test
    public void checkpoint() {
        int seconds = LocalTime.now().getSecond();
        Mono<Integer> source;
        if (seconds % 2 == 0) {
            source = Flux.range(1, 10)
                    .elementAt(5)
                    .checkpoint("source range(1,10)");
        } else if (seconds % 3 == 0) {
            source = Flux.range(0, 4)
                    .elementAt(5)
                    .checkpoint("source range(0,4)");
        } else {
            source = Flux.just(1, 2, 3, 4)
                    .elementAt(5)
                    .checkpoint("source just(1,2,3,4)");
        }

        source.block(); //line 186
    }

    @Test
    public void blockHound() {
        Flux.range(0, Runtime.getRuntime().availableProcessors() * 2)
                .log("first")
                .subscribeOn(Schedulers.parallel())
                .map(i -> {
                    CountDownLatch latch = new CountDownLatch(1);
                    Mono.delay(Duration.ofMillis(i * 100), Schedulers.boundedElastic())
                            .log("second")
                            .subscribe(it -> {
                                log.info("before count down");
                                latch.countDown();
                                log.info("after count down");
                            });
                    try {
                        log.info("before await" + i);
                        latch.await();
                        log.info("after await" + i);
                    } catch (InterruptedException e) {
                        throw new RuntimeException(e);
                    }

                    return i;
                })
                .blockLast();
    }

    @Test
    @SneakyThrows
    public void blockTest() {
        for (int i = 0; i < 15; i++) {
            Flux.interval(Duration.ofMillis(100))
                    .take(10)
                    .log("mono" + 1)
                    .subscribe(it -> log.info("onNext: {}", it));
        }
        Thread.sleep(300000);
    }


    @Test
    @SneakyThrows
    public void scheduler() {
        Flux.interval(Duration.ofMillis(1))
                .log("flux" + 3)
                .publishOn(Schedulers.parallel())
                .doOnNext(i -> {
                    try {
                        log.info("I'm here");
                        Thread.sleep(1);
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }
                })
                .subscribe(it -> log.info("handling: {}", it),
                        t -> log.error("handling error" + t.getMessage(), t));
        Thread.sleep(3000);
    }
}
